package advanced.design_mode.observermode;


import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.Flow;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.SubmissionPublisher;
import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;

/**
 * 利用Java 的提供的接口来做观察者模式
 */
public class MagazineSubscriber implements Flow.Subscriber<Integer> {
    public static final String JACK = "Jack";
    public static final String PETE = "Pete";
    private final long sleepTime;
    private final String subscriberName;
    private Flow.Subscription subscription;
    private int nextMagazineExpected;
    private int totalRead;

    public MagazineSubscriber(final long sleepTime, final String subscriberName) {
        this.sleepTime = sleepTime;
        this.subscriberName = subscriberName;
        this.nextMagazineExpected = 1;
        this.totalRead = 0;
    }

    /**
     * Publisher在被指定一个新的Subscriber时调用此方法。
     * 一般来说你需要在subscriber内部保存这个subscrition实例，
     * 因为后面会需要通过她向publisher发送信号来完成：请求更多数据，
     * 或者取消订阅。 一般在这里我们会直接请求第一个数据，正如代码中所示。
     * <p>
     * 作者：资良_Leo
     * 链接：https://www.jianshu.com/p/ce207be13078
     * 来源：简书
     * 简书著作权归作者所有，任何形式的转载都请联系作者获得授权并注明出处。
     *
     * @param subscription
     */
    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        this.subscription = subscription;
        subscription.request(1);
    }

    /**
     * 每当新的数据产生，这个方法会被调用
     *
     * @param magazineNumber
     */
    @Override
    public void onNext(Integer magazineNumber) {
        if (magazineNumber != nextMagazineExpected) {
            IntStream.range(nextMagazineExpected, magazineNumber).forEach(
                    (msgNumber) ->
                            log("Oh no! I missed the magazine " + msgNumber)
            );
            // Catch up with the number to keep tracking missing ones
            nextMagazineExpected = magazineNumber;
        }
        log("Great! I got a new magazine: " + magazineNumber);
        takeSomeRest();
        nextMagazineExpected++;
        totalRead++;

        log("I'll get another magazine now, next one should be: " +
                nextMagazineExpected);
        subscription.request(1);//请求一条数据
    }


    @Override
    public void onError(Throwable throwable) {
        System.out.println(" something has wrong");
    }

    /**
     * 当publisher数据推送完毕时会调用此方法，于是整个订阅过程结束。
     */
    @Override
    public void onComplete() {
        System.out.println("handRequest complete");
    }

    private void takeSomeRest() {
        try {
            Thread.sleep(sleepTime);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    public String getSubscriberName() {
        return subscriberName;
    }

    private void log(final String logMessage) {
        System.out.println("<=========== [" + subscriberName + "] : " + logMessage);
    }
}
